from typing import Dict, List, Tuple

import numpy as np
import pandas as pd

from ..models.data_source import CompareConfig
from ..utils.logger import logger
from .data_loader import DataLoader


class DataComparator:
    """数据对比器"""

    def __init__(self, config: CompareConfig):
        self.config = config

    def compare(self) -> pd.DataFrame:
        """执行数据对比"""
        try:
            # 加载数据
            loader1 = DataLoader(
                self.config.source1.type,
                self.config.source1.path,
                self.config.source1.db_config,
            )
            loader2 = DataLoader(
                self.config.source2.type,
                self.config.source2.path,
                self.config.source2.db_config,
            )

            df1 = loader1.load_data()
            df2 = loader2.load_data()

            # 打印数据源信息
            logger.info(
                f"数据源1信息:\n"
                f"类型: {self.config.source1.type}\n"
                f"路径: {self.config.source1.path}\n"
                f"映射: {self.config.source1.field_mapping}\n"
                f"数据行数: {len(df1)}"
            )

            logger.info(
                f"数据源2信息:\n"
                f"类型: {self.config.source2.type}\n"
                f"路径: {self.config.source2.path}\n"
                f"映射: {self.config.source2.field_mapping}\n"
                f"数据行数: {len(df2)}"
            )

            # 检查数据是否为空
            if df1.empty or df2.empty:
                logger.error("数据源为空")
                return pd.DataFrame()

            # 重命名列以区分来源
            df1.columns = [col.split("-")[0] for col in df1.columns]
            df2.columns = [col.split("-")[0] for col in df2.columns]

            # 应用字段映射
            df1 = df1.rename(columns=self.config.source1.field_mapping)
            df2 = df2.rename(columns=self.config.source2.field_mapping)

            # 打印映射后的信息
            logger.info(
                f"映射后的列名:\n"
                f"数据源1: {df1.columns.tolist()}\n"
                f"数据源2: {df2.columns.tolist()}\n"
                f"ID字段: {self.config.id_fields}"
            )

            # 检查并转换数值类型
            for col in df1.columns:
                if col in df2.columns:
                    # 检查两个数据源的同一列
                    values = pd.concat([df1[col], df2[col]]).dropna()
                    try:
                        # 检查是否所有值都可以转换为整数
                        if (
                            values.astype(float)
                            .apply(lambda x: float(x).is_integer())
                            .all()
                        ):
                            df1[col] = pd.to_numeric(df1[col], errors="coerce").astype(
                                "Int64"
                            )
                            df2[col] = pd.to_numeric(df2[col], errors="coerce").astype(
                                "Int64"
                            )
                            logger.info(f"列 {col} 已转换为整数类型")
                    except:
                        pass

            # 获取ID列名（带后缀）
            id_cols1 = [f"{col}-数据源1" for col in self.config.id_fields]
            id_cols2 = [f"{col}-数据源2" for col in self.config.id_fields]

            # 合并数据
            merged_df = pd.merge(
                df1,
                df2,
                on=self.config.id_fields,
                how="outer",
                suffixes=("-数据源1", "-数据源2"),
            )

            # 打印合并结果
            logger.info(
                f"合并后的数据信息:\n"
                f"行数: {len(merged_df)}\n"
                f"列: {merged_df.columns.tolist()}\n"
                f"前5行数据:\n{merged_df.head()}"
            )

            # 处理空值
            for col in merged_df.columns:
                if pd.api.types.is_numeric_dtype(merged_df[col]):
                    # 先转换为字符串，再处理空值
                    merged_df[col] = merged_df[col].astype(str).replace("nan", "")
                else:
                    merged_df[col] = merged_df[col].fillna("")

            # 重新排序列，确保所有字段都被包含
            ordered_columns = []

            # 首先添加ID列
            for id_field in self.config.id_fields:
                ordered_columns.append(f"{id_field}")  # 不添加后缀，因为这是合并的键列

            # 然后添加其他字段的数据源1和数据源2列
            other_fields = [
                f for f in self.config.fields if f["code"] not in self.config.id_fields
            ]
            for field in other_fields:
                field_code = field["code"]
                ordered_columns.extend(
                    [f"{field_code}-数据源1", f"{field_code}-数据源2"]
                )

            # 确保所有列都存在
            final_columns = [col for col in ordered_columns if col in merged_df.columns]

            # 重新排序列
            merged_df = merged_df[final_columns]

            # 如果不显示全部数据，则只显示有差异的数据
            if not self.config.show_all:
                # 检查每一行是否有差异
                has_diff = pd.Series(False, index=merged_df.index)
                for field in self.config.fields:
                    if field["code"] not in self.config.id_fields:
                        col1 = f"{field['code']}-数据源1"
                        col2 = f"{field['code']}-数据源2"
                        if col1 in merged_df.columns and col2 in merged_df.columns:
                            has_diff |= merged_df[col1] != merged_df[col2]
                        elif col1 in merged_df.columns or col2 in merged_df.columns:
                            # 如果只有一个数据源有这个字段，也认为是差异
                            has_diff |= True

                merged_df = merged_df[has_diff]

            logger.info("数据对比完成")
            return merged_df

        except Exception as e:
            logger.error(f"数据对比失败: {str(e)}", exc_info=True)
            raise

    def _filter_differences(self, df: pd.DataFrame) -> pd.DataFrame:
        """筛选出有差异的数据行"""
        # 获取非ID字段
        common_fields = set(df.columns) - set(self.config.id_fields)
        diff_mask = pd.Series(False, index=df.index)  # 创建一个布尔序列

        for field in common_fields:
            if field.endswith("-数据源1"):
                field2 = field.replace("-数据源1", "-数据源2")
                if field2 in df.columns:  # 确保对应的字段2存在
                    # 使用 fillna 处理空值，避免 NaN 的比较问题
                    diff_mask |= df[field].fillna("") != df[field2].fillna("")

        return df[diff_mask]

    def _get_field_name(self, field_code: str) -> str:
        """根据字段代码获取字段名称"""
        # 遍历所有数据项定义，查找匹配的字段名称
        for field in self.config.fields:
            if field["code"] == field_code:
                return field["name"]
        return field_code  # 如果找不到对应的名称，返回原始代码
